package com.xiashitech.agent.grpc;

import com.xiashitech.agent.config.StartConfig;
import com.xiashitech.interfaces.grpc.exchange.CommunicationObject;
import com.xiashitech.agent.instrumentation.spring.controller.apicapture.utils.ApiSyncUtil;
import com.xiashitech.agent.utils.JsonUtil;
import com.xiashitech.interfaces.agent.dto.api.ApiRequest;
import com.xiashitech.interfaces.agent.dto.api.DesenseParam;
import com.xiashitech.interfaces.grpc.*;
import io.grpc.ManagedChannel;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.StreamObserver;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.UUID;

public class XsApiInfoClient {
    private static final XsApiInfoClient client = new XsApiInfoClient();

    private final ConcurrentHashMap<String, CommunicationObject<ApiRequest>> apiData = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, CommunicationObject<DesenseParam>> deneseData = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, CommunicationObject<ApiRequest>> paramsData = new ConcurrentHashMap<>();

    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
    private volatile boolean apiDataSchedulerStarted = false;
    private RPCApiInfoServiceGrpc.RPCApiInfoServiceStub requestStub = null;
    private final Object requestStubLock = new Object();

    public static XsApiInfoClient getClient() {
        return client;
    }

    /**
     * 存在多线程并发访问的场景，加锁实现线程安全
     * @param apiRequest
     */
    public void pushApisToOdm(ApiRequest apiRequest) {
        System.out.println(Thread.currentThread().getId() + ", pushApisToOdm enter, StartConfig.agentService: " + StartConfig.getParamValueFromKey(StartConfig.agentService));
        ManagedChannel channelRef = XsChannelManager.getManager().buildStubOfChannel("127.0.0.1", 9988);
        System.out.println(Thread.currentThread().getId() + ", pushApisToOdm, now channel identityHashCode: "+ System.identityHashCode(channelRef));

        UUID apiRequestUuid = UUID.randomUUID();
        String key = apiRequestUuid.toString();
        CommunicationObject<ApiRequest> apiReqData = new CommunicationObject<>();
        apiReqData.setExchangeUuid(key);
        apiReqData.setData(apiRequest);
        apiReqData.setSendTimeStamp(System.currentTimeMillis());
        apiData.put(key, apiReqData);
        System.out.println("T" + System.nanoTime() + ", add one: " + key + ", now apiData size: " + apiData.size());

        synchronized (requestStubLock) {
            if(requestStub == null) {
                requestStub = RPCApiInfoServiceGrpc.newStub(channelRef);
            }
            if(!apiDataSchedulerStarted) {
                apiDataSchedulerStarted = true;
                scheduledExecutorService.scheduleAtFixedRate(()->{
                    // System.out.println("T" + System.nanoTime() + ", schedule apiData size: " + apiData.size());
                    Iterator<Map.Entry<String, CommunicationObject<ApiRequest>>> apiDataSet = apiData.entrySet().iterator();
                    while (apiDataSet.hasNext()) {
                        Map.Entry<String, CommunicationObject<ApiRequest>> item = apiDataSet.next();
                        CommunicationObject<ApiRequest> sendValue = item.getValue();
                        long timediff = System.currentTimeMillis() - sendValue.getSendTimeStamp();
                        if( timediff > 24 * 60 * 60 * 1000) { // 如果时间已经超过一天，表示已经过期，丢弃，不再做重发
                            System.out.println("remove, apiData uuid: " + item.getKey() + ", timestamp: " + sendValue.getData());
                            apiDataSet.remove();
                        } else  if( timediff > 10 * 1000) { // 如果时间已经超过 10s，表示需要重发
                            System.out.println("resend, apiData uuid: " + item.getKey() + ", timestamp: " + sendValue.getData());
                            sendApiData(sendValue.getData(), item.getKey());
                        }
                    }

                    // System.out.println("T" + System.nanoTime() + ", schedule deneseData size: " + deneseData.size());
                    Iterator<Map.Entry<String, CommunicationObject<DesenseParam>>> denenseDataSet = deneseData.entrySet().iterator();
                    while (denenseDataSet.hasNext()) {
                        Map.Entry<String, CommunicationObject<DesenseParam>> item = denenseDataSet.next();
                        CommunicationObject<DesenseParam> sendValue = item.getValue();
                        long timediff = System.currentTimeMillis() - sendValue.getSendTimeStamp();
                        if( timediff > 24 * 60 * 60 * 1000) { // 如果时间已经超过一天，表示已经过期，丢弃，不再做重发
                            System.out.println("remove, deneseData uuid: " + item.getKey() + ", timestamp: " + sendValue.getData());
                            apiDataSet.remove();
                        } else  if( timediff > 10 * 1000) { // 如果时间已经超过 10s，表示需要重发
                            System.out.println("resend, deneseData uuid: " + item.getKey() + ", timestamp: " + sendValue.getData());
                            sendDeneseData(sendValue.getData(), item.getKey());
                        }
                    }

                    // System.out.println("T" + System.nanoTime() + ", schedule paramsData size: " + paramsData.size());
                    apiDataSet = paramsData.entrySet().iterator();
                    while (apiDataSet.hasNext()) {
                        Map.Entry<String, CommunicationObject<ApiRequest>> item = apiDataSet.next();
                        CommunicationObject<ApiRequest> sendValue = item.getValue();
                        long timediff = System.currentTimeMillis() - sendValue.getSendTimeStamp();
                        if( timediff > 24 * 60 * 60 * 1000) { // 如果时间已经超过一天，表示已经过期，丢弃，不再做重发
                            System.out.println("remove, paramsData uuid: " + item.getKey() + ", timestamp: " + sendValue.getData());
                            apiDataSet.remove();
                        } else  if( timediff > 10 * 1000) { // 如果时间已经超过 10s，表示需要重发
                            System.out.println("resend, paramsData uuid: " + item.getKey() + ", timestamp: " + sendValue.getData());
                            sendParamsData(sendValue.getData(), item.getKey());
                        }
                    }
                }, 5, 10, TimeUnit.SECONDS);
            }
        }

        sendApiData(apiRequest, key);
    }

    private void sendApiData(ApiRequest apiRequest, String requestUuid) {
        StreamObserver<RPCApiSyncRequest> soRequest = requestStub.grpcApiToOmdStream(new StreamObserver<RPCApiSyncResponse>() {
            @Override
            public void onNext(RPCApiSyncResponse value) {
                System.out.println(Thread.currentThread().getId() + ", pushApisToOdm A response onNext: " + value.getCode() + ", uuid: " + value.getRequestUuid());
                // value 获取 key
                String successKey = value.getRequestUuid();
                apiData.remove(successKey);
            }

            @Override
            public void onError(Throwable t) {
                System.out.println(Thread.currentThread().getId() + ", pushApisToOdm A response onError: " + t.getMessage());
            }

            @Override
            public void onCompleted() {
                System.out.println(Thread.currentThread().getId() + ", pushApisToOdm response onCompleted");
            }
        });

        RPCApiSyncRequest rpcRequest = RPCApiSyncRequest.newBuilder().setApisJson(JsonUtil.convertObjectToString(apiRequest)).setRequestUuid(requestUuid).build();
        System.out.println(Thread.currentThread().getId() + ", pushApisToOdm, soRequest: " + System.identityHashCode(soRequest) + " before onNext, isReady: " + ((ClientCallStreamObserver) soRequest).isReady());

        soRequest.onNext(rpcRequest);
        soRequest.onCompleted();

        System.out.println(Thread.currentThread().getId() + ", pushApisToOdm, finish push, onComplete, request, client: " + System.identityHashCode(client));
    }

    public void grpcPushDesenseServer(DesenseParam desenseRequest) {
        System.out.println(Thread.currentThread().getId() + ", grpcPushDesenseServer enter, StartConfig.agentService: " + StartConfig.getParamValueFromKey(StartConfig.agentService));
        ManagedChannel channelRef = XsChannelManager.getManager().buildStubOfChannel("127.0.0.1", 9988);
        System.out.println(Thread.currentThread().getId() + ", grpcPushDesenseServer, now channel identityHashCode: "+ System.identityHashCode(channelRef));

        UUID apiRequestUuid = UUID.randomUUID();
        String key = apiRequestUuid.toString();
        CommunicationObject<DesenseParam> apiReqData = new CommunicationObject<>();
        apiReqData.setExchangeUuid(key);
        apiReqData.setData(desenseRequest);
        apiReqData.setSendTimeStamp(System.currentTimeMillis());
        deneseData.put(key, apiReqData);
        System.out.println("T" + System.nanoTime() + ", add one: " + key + ", now deneseData size: " + apiData.size());

        synchronized (requestStubLock) {
            if(requestStub == null) {
                requestStub = RPCApiInfoServiceGrpc.newStub(channelRef);
            }
        }

        sendDeneseData(desenseRequest, key);
    }

    private void sendDeneseData(DesenseParam desenseRequest, String requestUuid) {
        StreamObserver<RPCDesenseParamRequest> soDesensePushRequest = requestStub.biSyncApiDesense(new StreamObserver<RPCDesenseParamResponse>() {
            @Override
            public void onNext(RPCDesenseParamResponse value) {
                System.out.println(Thread.currentThread().getId() + ", grpcPushDesenseServer A onNext: " + value.getCode() + ", uuid: " + value.getRequestUuid());
                // 不管返回啥，都表示服务端已收到，放到缓存，不再发送
                String successKey = value.getRequestUuid();
                deneseData.remove(successKey);
                // 更新业务缓存
                ApiSyncUtil.cacheDesense(desenseRequest);
            }

            @Override
            public void onError(Throwable t) {
                System.out.println(Thread.currentThread().getId() + ", grpcPushDesenseServer A onError: " + t.getMessage());
            }

            @Override
            public void onCompleted() {
                System.out.println(Thread.currentThread().getId() + ", grpcPushDesenseServer A onCompleted");
            }
        });

        RPCDesenseParamRequest rpcRequest = RPCDesenseParamRequest.newBuilder().setDesenseParamJson(JsonUtil.convertObjectToString(ApiRequest.builder().desenseParam(desenseRequest).build())).setRequestUuid(requestUuid).build();
        soDesensePushRequest.onNext(rpcRequest);
        soDesensePushRequest.onCompleted();
        System.out.println(Thread.currentThread().getId() + ", grpcPushDesenseServer, finish push, onComplete, soDesensePushRequest: " + System.identityHashCode(soDesensePushRequest));
    }

    public void grpcApiParasToOMD(ApiRequest apiRequest) {
        System.out.println(Thread.currentThread().getId() + ", grpcApiParasToOMD enter, StartConfig.agentService: " + StartConfig.getParamValueFromKey(StartConfig.agentService));
        ManagedChannel channelRef = XsChannelManager.getManager().buildStubOfChannel("127.0.0.1", 9988);
        System.out.println(Thread.currentThread().getId() + ", grpcApiParasToOMD, now channel identityHashCode: "+ System.identityHashCode(channelRef));

        UUID apiRequestUuid = UUID.randomUUID();
        String key = apiRequestUuid.toString();
        CommunicationObject<ApiRequest> apiReqData = new CommunicationObject<>();
        apiReqData.setExchangeUuid(key);
        apiReqData.setData(apiRequest);
        apiReqData.setSendTimeStamp(System.currentTimeMillis());
        paramsData.put(key, apiReqData);
        System.out.println("T" + System.nanoTime() + ", add one: " + key + ", now paramsData size: " + apiData.size());

        synchronized (requestStubLock) {
            if(requestStub == null ) {
                requestStub = RPCApiInfoServiceGrpc.newStub(channelRef);
            }
        }
        sendParamsData(apiRequest, key);
    }

    private void sendParamsData(ApiRequest apiRequest, String requestUuid) {
        StreamObserver<RPCApiSyncRequest> soApiParamsPushRequest = requestStub.grpcApiParamsToOmdStream(new StreamObserver<RPCApiSyncResponse>() {
            @Override
            public void onNext(RPCApiSyncResponse value) {
                System.out.println(Thread.currentThread().getId() + ", grpcApiParasToOMD A receive response onNext: " + value.getCode() + ", uuid: " + value.getRequestUuid());

                // 清除发送暂存
                String reqUuid = value.getRequestUuid();
                paramsData.remove(reqUuid);
            }

            @Override
            public void onError(Throwable t) {
                System.out.println(Thread.currentThread().getId() + ", grpcApiParasToOMD A response onError: " + t.getMessage());
            }

            @Override
            public void onCompleted() {
                System.out.println(Thread.currentThread().getId() + ", grpcApiParasToOMD A response onCompleted");
            }
        });
        RPCApiSyncRequest rpcRequest = RPCApiSyncRequest.newBuilder().setApisJson(JsonUtil.convertObjectToString(apiRequest)).setRequestUuid(requestUuid).build();

        soApiParamsPushRequest.onNext(rpcRequest);
        // 发送完，就可以结束流，下次调用会新开流
        soApiParamsPushRequest.onCompleted();

        System.out.println(Thread.currentThread().getId() + ", grpcApiParasToOMD, finish push, onComplete, soApiParamsPushRequest: " + System.identityHashCode(soApiParamsPushRequest));
    }
}
